package com.java.mall_imooc.common.mq;

import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.java.mall_imooc.common.exception.RRException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

/**
 * Rabbit 发送消息
 */
@Component
@Slf4j
public class RabbitSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback, InitializingBean {
    @Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")
    private int mqRetryCount;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送MQ消息,异步
     *
     * @param exchangeName 交换机名称
     * @param routingKey   路由名称
     * @param message      发送消息体
     */
    public void sendMessage(String exchangeName, String routingKey, String message) {
        Assert.notNull(message, "message 消息体不能为NULL");
        Assert.notNull(exchangeName, "exchangeName 不能为NULL");
        Assert.notNull(routingKey, "routingKey 不能为NULL");
        // 获取CorrelationData对象
        CorrelationData correlationData = correlationData(message);
        correlationData.setExchange(exchangeName);
        correlationData.setRoutingKey(routingKey);
        correlationData.setMessage(JSONUtil.toJsonStr(message));

        log.info("发送MQ消息，消息ID：{}，消息体:{}, exchangeName:{}, routingKey:{}",
                correlationData.getId(), JSONUtil.toJsonStr(message), exchangeName, routingKey);
        // 发送消息
        this.convertAndSend(exchangeName, routingKey, message, correlationData);
    }

    /**
     * RPC方式，发送MQ消息
     *
     * @param exchangeName 交换机名称
     * @param routingKey   路由名称
     * @param message      发送消息体
     */
    public void sendAndReceiveMessage(String exchangeName, String routingKey, String message) {
        Assert.notNull(message, "message 消息体不能为NULL");
        Assert.notNull(exchangeName, "exchangeName 不能为NULL");
        Assert.notNull(routingKey, "routingKey 不能为NULL");
        // 获取CorrelationData对象
        CorrelationData correlationData = this.correlationData(message);
        correlationData.setExchange(exchangeName);
        correlationData.setRoutingKey(routingKey);
        correlationData.setMessage(JSONUtil.toJsonStr(message));

        log.info("发送MQ消息，消息ID：{}，消息体:{}, exchangeName:{}, routingKey:{}",
                correlationData.getId(), JSONUtil.toJsonStr(message), exchangeName, routingKey);

        rabbitTemplate.convertSendAndReceive(exchangeName, routingKey, message);
    }

    /**
     * 用于实现消息发送到RabbitMQ交换器后接收ack回调。
     * 如果消息发送确认失败就进行重试。
     *
     * @param correlationData 消息相关数据对象
     * @param ack             ack应答
     * @param cause           失败原因
     */
    @Override
    public void confirm(org.springframework.amqp.rabbit.connection.CorrelationData correlationData, boolean ack, String cause) {
        CorrelationData correlationDataExtends = null;
        if (correlationData instanceof CorrelationData) {
            correlationDataExtends = (CorrelationData) correlationData;
        }

        // 消息回调确认失败处理
        if (!ack) {
            if (correlationDataExtends != null) {
                //消息发送失败,就进行重试，重试过后还不能成功就记录到数据库
                if (correlationDataExtends.getRetryCount() < mqRetryCount) {
                    log.info("MQ消息发送失败，消息重发，消息ID：{}，重发次数：{}，消息体:{}", correlationDataExtends.getId(),
                            correlationDataExtends.getRetryCount(), correlationDataExtends.getMessage());

                    // 将重试次数+1
                    correlationDataExtends.setRetryCount(correlationDataExtends.getRetryCount() + 1);

                    // 重发发消息
                    this.convertAndSend(correlationDataExtends.getExchange(), correlationDataExtends.getRoutingKey(),
                            correlationDataExtends.getMessage(), correlationDataExtends);
                } else {
                    //消息重新发送失败,将消息放到数据库等待补发
                    log.error("MQ消息重发失败，消息ID：{}，消息体:{}", correlationData.getId(), correlationDataExtends.getMessage());
                    // TODO 保存消息到数据库
                }
            }
        }
        log.info("消息发送成功,消息ID:{}", correlationData.getId());
    }

    /**
     * 用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
     * 在脑裂的情况下会出现这种情况。
     */
    @Override
    public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("MQ路由消息失败，replyCode:{}, replyText:{}，exchange:{}，routingKey:{}，消息体:{}",
                replyCode, replyText, exchange, routingKey, JSONUtil.toJsonStr(message.getBody()));

        // TODO 保存消息到数据库
        throw new RRException("MQ路由消息失败");
    }

    /**
     * 消息相关数据（消息ID）
     *
     * @param message   消息体
     */
    private CorrelationData correlationData(String message) {
        // 消息ID默认使用UUID
        return new CorrelationData(IdUtil.simpleUUID(), message);
    }

    /**
     * 发送消息
     *
     * @param exchange        交换机名称
     * @param routingKey      路由key
     * @param message         消息内容
     * @param correlationData 消息相关数据（消息ID）
     */
    private void convertAndSend(String exchange, String routingKey, final Object message, CorrelationData correlationData) {
        try {
            rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
        } catch (Exception e) {
            log.error("MQ消息发送异常，消息ID：{}，消息体:{}, exchangeName:{}, routingKey:{}",
                    correlationData.getId(), JSONUtil.toJsonStr(message), exchange, routingKey, e);

            throw new RRException("MQ消息发送异常");
        }
    }

    @Override
    public void afterPropertiesSet() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

}